package com.mng.window;

import com.mng.dto.SensorReading;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class processWindowTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> stringDs = env.socketTextStream("localhost",7777);
        SingleOutputStreamOperator<SensorReading> sensorDs = stringDs.map(line ->{
                    String[] strings = line.split(",");
                    return new SensorReading(strings[0],Long.parseLong(strings[1]),Double.parseDouble(strings[2]));
                }
        );
        sensorDs.keyBy(sensor -> sensor.getId())
                .window(SlidingProcessingTimeWindows.of(Time.seconds(30),Time.seconds(10)))
                .process(new ProcessWindowFunction<SensorReading, SensorReading, String, TimeWindow>() {
                    @Override
                    public void clear(Context context) throws Exception {
                        super.clear(context);
                    }

                    @Override
                    public void process(String s, Context context, Iterable<SensorReading> iterable, Collector<SensorReading> collector) throws Exception {
                        Long maxTime = 0L;
                        Double sumTem = 0D;
                        Long count = 0L;
                        for(SensorReading sensorReading  : iterable){
                            if(sensorReading.getTimestamp() > maxTime){
                                maxTime = sensorReading.getTimestamp();
                            }
                            sumTem += sensorReading.getTemperaturre();
                            count +=1;
                        }
                        collector.collect(new SensorReading(s,maxTime,sumTem/count));
                    }
                })
                .print();
        env.execute();
    }
}
